package com.JDK.Thread;


import com.JDK.ThreadPoolExecutor.*;
import org.junit.Test;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 测试线程池
 */
public class ThreadPoolTest {


    AtomicInteger inc = new AtomicInteger(0);


    MyThreadPoolExecutor executor = new MyThreadPoolExecutor(
            8,
            16,
            3L,
            TimeUnit.SECONDS,
            new FixArrayBlockingQueue<>(50),
            new MyThreadPoolExecutor.AbortPolicy(),
            50L
    );

    /**
     * 测试 ThreadPoolExecutor
     */
    @Test
    public void test1() throws ExecutionException, InterruptedException {

        Callable<Integer> c = () -> {
            Thread.sleep(100);
            return inc.incrementAndGet();
        };

        HashSet<Future> futureSet = new HashSet<>();

        for (int i = 0; i < 100; i++) {
            MyTask myTask = new MyTask(inc, i);
            try {
                Thread.sleep(2);
                Future<Result> future = executor.submit(myTask);
                futureSet.add(future);
            } catch (RejectedExecutionException ex) {
                System.out.println("无法再容纳更多的任务，任务 " + i + " 已被拒绝执行");

            }


            System.out.println("第 " + i + " 个任务  queue size: " + executor.getQueue().size()
                    + " workerCount: " + executor.getWorkerCount());

        }

        int finishCount = 0;

        for (Future future : futureSet) {
            try {
                if (future.get() != null) {
                    finishCount++;
                    System.out.println(future.isDone() + " " + future.get());
                }

            } catch (Exception ex) {
                System.out.println(ex.getMessage());

            }
        }

        System.out.println("Finish Count: " + finishCount);

        System.out.println("core: " + executor.getWorkerCount());

        System.out.println(Integer.toBinaryString(executor.ctl.get()));

        Thread.sleep(5000);

        System.out.println("core: " + executor.getWorkerCount());

        System.out.println(Integer.toBinaryString(executor.ctl.get()));

        executor.shutdown();
    }

    /**
     * 测试 Future
     */
    @Test
    public void testFuture() throws ExecutionException, InterruptedException {
        HashSet<Future> futureSet = new HashSet<>();
        for (int i = 0; i < 100; i++) {
            MyTask myTask = new MyTask(inc, i);
            try {
                Thread.sleep(5);
                Future<Result> future = executor.submit(myTask);
                futureSet.add(future);
            } catch (RejectedExecutionException ex) {
                System.out.println("无法再容纳更多的任务，任务 " + i + " 已被拒绝执行");

            }

            System.out.println("第 " + i + " 个任务  queue size: " + executor.getQueue().size()
                    + " workerCount: " + executor.getWorkerCount());
        }


        int finishCount = 0;

        for (Future future : futureSet) {
            try {
                if (future.get() != null) {
                    finishCount++;
                    System.out.println(future.isDone() + " " + future.get());
                }

            } catch (Exception ex) {
                System.out.println(ex.getMessage());

            }
        }

        System.out.println("Finish Count: " + finishCount);

        executor.shutdown();

    }

    /**
     * 测试工作线程的状态
     */
    @Test
    public void testWorkerState() throws ExecutionException, InterruptedException {
        ArrayList<MyThreadPoolExecutor.Worker> workers;
        HashSet<Future> futureSet = new HashSet<>();
        for (int i = 0; i < 100; i++) {
            MyTask myTask = new MyTask(inc, i);
            try {
                Thread.sleep(2);
                Future<Result> future = executor.submit(myTask);
                futureSet.add(future);
            } catch (RejectedExecutionException ex) {
                System.out.println("无法再容纳更多的任务，任务 " + i + " 已被拒绝执行");

            }

//            System.out.println("第 " + i + " 个任务  queue size: " + executor.getQueue().size()
//                    + " workerCount: " + executor.getWorkerCount());
        }

        for (Future future : futureSet) {
            System.out.println(future.get());
        }


        System.out.println("================= workers ================");

        workers = executor.getWorkers();

        for (MyThreadPoolExecutor.Worker worker : workers) {
            System.out.println("workerId: " + worker.getId() + ", complete task: " + worker.getCompletedTasks() +
                    ", worker is lock? : " + worker.isLocked());
        }

        System.out.println("================= after keep alive time ================");
        Thread.sleep(4000);

        workers = executor.getWorkers();

        for (MyThreadPoolExecutor.Worker worker : workers) {
            System.out.println("workerId: " + worker.getId() + ", complete task: " + worker.getCompletedTasks() +
                    ", worker is lock? : " + worker.isLocked());
        }


    }

    /**
     * 测试线程池的 Hash 路由分发任务
     */
    @Test
    public void testHashRoute(){

    }

    @Test
    public void testChange(){
        FixArrayBlockingQueue blockingQueue = (FixArrayBlockingQueue)executor.getQueue();
        System.out.println(blockingQueue.getCapacity());
        blockingQueue.setCapacity(100);
        System.out.println(blockingQueue.getCapacity());
    }


    /**
     *  统计每秒的任务完成率
     */
    @Test
    public void testCountSpeed() throws InterruptedException, ExecutionException {
        HashSet<Future> futureSet = new HashSet<>();

        long preCount = 0;
        long curCount = 0;

        for (int i = 1; i <= 100; i++) {
            MyTask myTask = new MyTask(inc, i);
            try {
                Thread.sleep(5 );
                Future<Result> future = executor.submit(myTask);
                futureSet.add(future);
            } catch (RejectedExecutionException ex) {
                System.out.println("无法再容纳更多的任务，任务 " + i + " 已被拒绝执行");
            }

            if(i % 10 == 0){
                System.out.println("queue task count: " + executor.getQueue().size());

                curCount = executor.getCompletedTaskCount();
                System.out.println("speed: " + (curCount - preCount) * 20 + " task/s");
                preCount = curCount;

                System.out.println("worker count: " + executor.getWorkerCount() +
                        " ative count: " + executor.getActiveCount());
            }

        }

        for (Future future : futureSet) {
            Result res = (Result) future.get();
            System.out.println(res);
        }


        System.out.println("all task count: " + executor.getAllTaskCount());
        System.out.println("all complete task count: " + executor.getCompletedTaskCount());
        System.out.println("all reject task count: " + executor.getRejectCount());

        System.out.println("finish rate: " +
                executor.getCompletedTaskCount() * 100.0 / (double)executor.getAllTaskCount() + " %");

        System.out.println("reject rate: " +
                executor.getRejectCount() * 100.0 / (double)executor.getAllTaskCount() + " %");

        System.out.println("task avg run time: " + executor.getTaskAvgFinishTime() + " ms");

        System.out.println("time out task count: " + executor.getTaskTimeOutCount());



    }


    /**
     * 判断是否含有重复的结果
     */
    public boolean hasDupRes(HashSet<Result> resultSet){
        HashSet<Integer> set = new HashSet<>();
        for (Result res : resultSet) {
            set.add(res.res);
        }
        return set.size() != resultSet.size();
    }

}
